package handler

import (
	"context"
	"sync"
	"time"

	"github.com/go-redis/redis/v8"
	consulapi "github.com/hashicorp/consul/api"
	"github.com/hashicorp/consul/api/watch"
	"go.uber.org/zap"
	"google.golang.org/grpc/codes"
	"google.golang.org/grpc/status"

	proto "mxshop_srvs/consul_srv/proto/gen/v1"
)

func (s *Service) GetWorkerId(c context.Context, req *proto.ServiceEntity) (*proto.WorkerEntity, error) {
	workerID, err := s.RDB.GetWorkerID(c, req.ServiceId) // 获取workerID(优先从归还池里获取)
	if err != nil {
		if err != redis.Nil {
			return nil, status.Error(codes.Internal, "")
		}
	}
	if workerID != 0 {
		err := s.RDB.RecordWorkerID(c, req.ServiceId, workerID) // 如果成功获取则做个记录
		if err != nil {
			return nil, status.Error(codes.Internal, "")
		}
		return &proto.WorkerEntity{WorkerId: workerID}, nil
	}
	workerID, err = s.RDB.NewWorkerID(c, req.ServiceId) // 为获取则使用Incr去新建一个workerID
	if err != nil {
		return nil, status.Error(codes.Internal, "")
	}
	return &proto.WorkerEntity{WorkerId: workerID}, nil
}

func (s *Service) ReturnWorkerId(c context.Context, req *proto.ServiceEntity) (*proto.ReturnWorkerIdResponse, error) {
	workerID, err := s.RDB.GetWorkerID(c, req.ServiceId)
	if err != nil {
		return nil, status.Error(codes.Internal, "")
	}
	err = s.RDB.ReturnWorkerID(c, req.ServiceId, workerID)
	if err != nil {
		return nil, status.Error(codes.Internal, "")
	}
	return &proto.ReturnWorkerIdResponse{}, nil
}

// watch包的使用方法为：1）使用watch.Parse(查询参数)生成Plan，2）绑定Plan的handler，3）运行Plan

// Watcher 定义watcher
type Watcher struct {
	Wp       *watch.Plan            // 总的Services变化对应的Plan
	watchers map[string]*watch.Plan // 对已经进行监控的service作个记录
	RWMutex  *sync.RWMutex
}

func (s *Service) Handler(serviceID string) {
	<-time.After(25 * time.Second) // 等待25s执行:健康检查5s+撤销服务时间15s+5s缓冲
	s.Watcher.RWMutex.RLock()
	defer s.Watcher.RWMutex.RUnlock()

	ctx := context.Background()
	ok, err := s.RDB.GetTag(ctx, serviceID)
	if err != nil {
		zap.L().Error("failed to get tag", zap.Error(err))
		return
	}
	if !ok {
		zap.L().Info("no service tag", zap.String("service_id", serviceID))
		return
	}

	err = s.RDB.DelTag(ctx, serviceID)
	workerID, err := s.RDB.GetWorkerID(ctx, serviceID)
	if err != nil {
		zap.L().Error("failed to get worker id", zap.String("service_id", serviceID), zap.Error(err))
		return
	}
	err = s.RDB.ReturnWorkerID(ctx, serviceID, workerID)
	if err != nil {
		zap.L().Error(
			"failed to return worker id",
			zap.String("service_id", serviceID),
			zap.Int64("worker_id", workerID),
			zap.Error(err),
		)
		return
	}
	zap.L().Info("return worker id success", zap.String("service_id", serviceID))
}

// 将consul新增的service加入，并监控
func (s *Service) registerServiceWatcher(serviceName, address string) error {
	// watch endpoint 的请求参数，具体见官方文档：https://www.consul.io/docs/dynamic-app-config/watches#service
	wp, err := watch.Parse(map[string]interface{}{
		"type":    "service",
		"service": serviceName,
	})
	if err != nil {
		return err
	}

	// 定义service变化后所执行的程序(函数)handler
	wp.Handler = func(idx uint64, data interface{}) {
		switch d := data.(type) {
		case []*consulapi.ServiceEntry:
			for _, i := range d {
				// 这里是单个service变化时需要做的逻辑，可以自己添加，或在外部写一个类似handler的函数传进来
				// fmt.Printf("service %s 已变化", i.Service.ID)
				// 打印service的状态
				// fmt.Println("service status: ", i.Checks.AggregatedStatus())

				ctx := context.Background()
				zap.L().Info("service change", zap.String(i.Service.ID, i.Checks.AggregatedStatus()))
				switch i.Checks.AggregatedStatus() {
				case consulapi.HealthCritical: // 健康检查出错
					s.Watcher.RWMutex.Lock()
					err = s.RDB.AddTag(ctx, i.Service.ID)
					if err != nil {
						zap.L().Error("failed to add service tag", zap.Error(err))
					}
					s.Watcher.RWMutex.Unlock()
					go s.Handler(i.Service.ID)
				case consulapi.HealthPassing: // 健康检查通过
					s.Watcher.RWMutex.Lock()
					err = s.RDB.DelTag(ctx, i.Service.ID)
					if err != nil {
						zap.L().Error("failed to del service tag", zap.Error(err))
					}
					s.Watcher.RWMutex.Unlock()
				}
			}
		}
	}
	// 启动监控
	go func(address string) {
		_ = wp.Run(address)
	}(address)
	// 对已启动监控的service作一个记录
	s.Watcher.RWMutex.Lock()
	s.Watcher.watchers[serviceName] = wp
	s.Watcher.RWMutex.Unlock()

	return nil
}

func (s *Service) NewWatcher(watchType string, opts map[string]string, consulAddr string) (*Watcher, error) {
	var options = map[string]interface{}{
		"type": watchType,
	}
	// 组装请求参数。(监控类型不同，其请求参数不同)
	for k, v := range opts {
		options[k] = v
	}

	wp, err := watch.Parse(options)
	if err != nil {
		return nil, err
	}

	w := &Watcher{
		Wp:       wp,
		watchers: make(map[string]*watch.Plan),
		RWMutex:  new(sync.RWMutex),
	}

	wp.Handler = func(idx uint64, data interface{}) {
		switch d := data.(type) {
		// 这里只实现了对services的监控，其他监控的data类型判断参考：https://github.com/dmcsorley/avast/blob/master/consul.go
		// services
		case map[string][]string:
			for i := range d {
				// 如果该service已经加入到ConsulRegistry的services里监控了，就不再加入 或者i 为 "consul"的字符串
				// 为什么会多一个consul，参考官方文档services监听的返回值：https://www.consul.io/docs/dynamic-app-config/watches#services
				if _, ok := w.watchers[i]; ok || i == "consul" {
					continue
				}
				_ = s.registerServiceWatcher(i, consulAddr)
			}

			// 从总的services变化中找到不再监控的service并停止
			s.Watcher.RWMutex.RLock()
			watches := s.Watcher.watchers
			s.Watcher.RWMutex.RUnlock()

			// remove unknown services from watchers
			for i, svc := range watches {
				if _, ok := d[i]; !ok {
					svc.Stop()
					delete(watches, i)
				}
			}
		default:
			zap.L().Warn("不能判断监控的数据类型", zap.Any("plan", &d))
		}
	}
	return w, nil
}

func (s *Service) RegisterWatcher(watchType string, opts map[string]string, consulAddr string) error {
	var err error
	s.Watcher, err = s.NewWatcher(watchType, opts, consulAddr)
	if err != nil {
		return err
	}
	defer s.Watcher.Wp.Stop()
	if err = s.Watcher.Wp.Run(consulAddr); err != nil {
		return err
	}
	return nil
}
